package cn.kgc.gmall.app.dwd;

import cn.kgc.gmall.app.func.DimSink;
import cn.kgc.gmall.app.func.MyDeserializationSchemaFunction;
import cn.kgc.gmall.app.func.TableProcessFunction;
import cn.kgc.gmall.bean.TableProcess;
import cn.kgc.gmall.utils.MyKafkaUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * 读入kafka中ods层的业务数据  进行简单过滤 使用flinkcdc技术动态读取MySQL中的配置表信息
 *
 * 使用双流操作  存入list状态变量中  进行动态的分流操作
 *
 */
public class BaseDBApp {

    public static void main(String[] args) throws Exception {

        // 1.创建flink的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2.设置并行度
        env.setParallelism(4);

        // 3.设置检查点 精准一次
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 4.检查点启动时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // 4.1 如果flink程序关闭是否删除检查点
        env.getCheckpointConfig()
                .enableExternalizedCheckpoints(
                        CheckpointConfig
                                .ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
                );
        // 5.状态后端保存检查点到hdfs上
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/flink/checkpoint"));
        // 6.设置用户
        System.setProperty("HADOOP_USER_NAME","atkgc");
        String topic = "ods_base_db_m";
        String groupId = "basedbapp_group";
        // 7.读取kafka中ods_base_db_m中的数据
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtils.getKafkaConsumerSource(topic, groupId));

        // 7.1 将String 类型的数据转化为json对象
        /*kafkaDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {
                return JSON.parseObject(value);
            }
        })*/
        // kafkaDS.map(x -> JSON.parseObject(x))
        SingleOutputStreamOperator<JSONObject> mapDS = kafkaDS.map(JSON::parseObject);

        // 8.对这些数据进行简单ETL
        SingleOutputStreamOperator<JSONObject> filterDS = mapDS.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {

                // 如果没有表名  如果数据小于3条
                return  value.getString("table") != null &&
                        value.getString("table").length() > 0 &&
                        value.getJSONObject("data") != null &&
                        // 字段小于3的表都会被过滤掉所以调整为大于1
                        value.getJSONObject("data").size() > 1;
            }
        });
        filterDS.print("输出主流数据:");
        // 展示过滤的数据
        //filterDS.print();
        // 9.使用flinkCDC读取MySQL中配置表中的数据
        // 使用数据流的方式
        SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                // 配置你要导入的数据库 可以写多个数据库
                .databaseList("gmall_2022_realtime")
                // 配置你要导入的表
                .tableList("gmall_2022_realtime.table_process")
                .username("root")
                .password("root")
                // 启动选项 initial:启动的时候扫描全表将所有的数据扫描出来再监控binlog文件
                // earliest: 启动的时候只加载binlog中的数据
                // 利用检查点保留机制
                .startupOptions(StartupOptions.initial())
                // 使用自己定义的序列化器
                .deserializer(new MyDeserializationSchemaFunction())
                .build();

        DataStreamSource<String> tableDS = env.addSource(sourceFunction);
        //tableDS.print();

        // 10.进行动态分流
        // filterDS 四个插槽  tableDS: 定义为广播变量
        // 状态变量一个 map descriptor，它描述了用于存储规则名称与规则本身的 map 存储结构
        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>(
                "table-process",
                String.class,
                TableProcess.class
                );

        // 广播流，广播规则并且创建 broadcast state
        /* BroadcastStream<Rule> ruleBroadcastStream = ruleStream
                .broadcast(ruleStateDescriptor); */
        // 将tableDs使用广播变量广播到每一个插槽中
        BroadcastStream<String> broadcastStream = tableDS.broadcast(mapStateDescriptor);

        // 连接broadcastStream 和 filterDS
        BroadcastConnectedStream<JSONObject, String> connectedStream = filterDS.connect(broadcastStream);

        // 将双流中的数据进行匹配  普通流 和广播流相连 使用的是 BroadcastProcessFunction
        // 处理函数里面 将事实表输出到主流  kafka   维度表 输出到侧输出流  hbase
        OutputTag<JSONObject> outputTag = new OutputTag<JSONObject>("hbase"){};
        // 将侧输出流 和 状态变量 传入到tableProcessFunction
        SingleOutputStreamOperator<JSONObject> processStream = connectedStream.process(new TableProcessFunction(outputTag,mapStateDescriptor));

        // 使用侧输出流 将hbase相关的通过phoenix写道hbase中
        DataStream<JSONObject> sideOutput = processStream.getSideOutput(outputTag);

        sideOutput.addSink(new DimSink() );
        // 主流的数据写出到kafka中 dws层
        processStream.addSink(MyKafkaUtils.getKafkaProducerSink(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {

                // 1.从配置表中拿主题名
                String sink_table = jsonObj.getString("sink_table");
                String data = jsonObj.getString("data");
                return new ProducerRecord<>(sink_table,data.getBytes());
            }
        }));
        env.execute();

    }


}
